package hadoop;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.CompareFilter;
import org.apache.hadoop.hbase.filter.RegexStringComparator;
import org.apache.hadoop.hbase.filter.RowFilter;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;

public class FinalHbase {
    public static void main(String[] args) throws IOException {
        System.out.println("此功能完成对Hbase数据库表的操作");
        //scanTable("lijie0315:lijie_forum");
        //insertData("lijie0315:user");
        UserFilter("lijie0315:user");
    }
    public static Connection ConnHbase(){

        Configuration conf= HBaseConfiguration.create();

        conf.set("hbase.zookeeper.quorum","192.168.31.131:2181,192.168.31.134:2181,192.168.31.133:2181");
        Connection con=null;

        try {
            con= ConnectionFactory.createConnection(conf);
            System.out.println("连接Hbase成功");
        } catch (IOException e) {
            System.out.println("连接Hbase数据库报错，具体信息如："+e.getMessage());
            throw new RuntimeException(e);
        }
        return con;
    }
    public static void scanTable(String tableName) {
        try {
            Connection conn = ConnHbase();
            Table table = conn.getTable(TableName.valueOf(tableName));
            Scan scan = new Scan();
            scan.setMaxResultSize(5);
            ResultScanner scanner = table.getScanner(scan);
            for (Result result : scanner) {
                for (Cell cell : result.listCells()) {
                    String rowKey = Bytes.toString(CellUtil.cloneRow(cell));
                    String family = Bytes.toString(CellUtil.cloneFamily(cell));
                    String qualifier = Bytes.toString(CellUtil.cloneQualifier(cell));
                    String value = Bytes.toString(CellUtil.cloneValue(cell));
                    System.out.println("编码为: " + rowKey + ", Family: " + family + ", Qualifier: " + qualifier + ", Value: " + value);
                }
            }
            table.close();
            conn.close();
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }
    public static boolean insertData(String tableName) {
        boolean str = false;
        try {
            Connection conn = ConnHbase();
            Table table = conn.getTable(TableName.valueOf(tableName));
            // 插入数据
            Put put1 = new Put(Bytes.toBytes("lijie"));
            put1.addColumn(Bytes.toBytes("usrinfo"), Bytes.toBytes("name"), Bytes.toBytes("lijie"));
            put1.addColumn(Bytes.toBytes("usrinfo"), Bytes.toBytes("sex"), Bytes.toBytes("male"));
            put1.addColumn(Bytes.toBytes("usrinfo"), Bytes.toBytes("age"), Bytes.toBytes("21"));

            Put put2 = new Put(Bytes.toBytes("user2"));
            put2.addColumn(Bytes.toBytes("usrinfo"), Bytes.toBytes("name"), Bytes.toBytes("chenxixi"));
            put2.addColumn(Bytes.toBytes("usrinfo"), Bytes.toBytes("sex"), Bytes.toBytes("male"));
            put2.addColumn(Bytes.toBytes("usrinfo"), Bytes.toBytes("age"), Bytes.toBytes("22"));

            Put put3 = new Put(Bytes.toBytes("user3"));
            put3.addColumn(Bytes.toBytes("usrinfo"), Bytes.toBytes("name"), Bytes.toBytes("zhangsan"));
            put3.addColumn(Bytes.toBytes("usrinfo"), Bytes.toBytes("sex"), Bytes.toBytes("female"));
            put3.addColumn(Bytes.toBytes("usrinfo"), Bytes.toBytes("age"), Bytes.toBytes("23"));

            table.put(put1);
            table.put(put2);
            table.put(put3);
            table.close();
            conn.close();
            System.out.println("数据插入成功");
            str = true;
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
        return str;
    }

    public static boolean UserFilter(String tableName) throws IOException {
        boolean str = false;
        try {
            Connection conn = ConnHbase();
            Table table = conn.getTable(TableName.valueOf(tableName));

        // 创建过滤器
        RowFilter rowFilter = new RowFilter(CompareFilter.CompareOp.EQUAL, new RegexStringComparator(".*lijie.*"));
        Scan scan = new Scan();
        scan.setFilter(rowFilter);
        // 执行查询并打印结果
        ResultScanner scanner = table.getScanner(scan);
        for (Result result : scanner) {
            String rowKey = Bytes.toString(result.getRow());
            String name = Bytes.toString(result.getValue(Bytes.toBytes("usrinfo"), Bytes.toBytes("name")));
            String sex = Bytes.toString(result.getValue(Bytes.toBytes("usrinfo"), Bytes.toBytes("sex")));
            String age = Bytes.toString(result.getValue(Bytes.toBytes("usrinfo"), Bytes.toBytes("age")));
            System.out.printf("RowKey: %s, Name: %s, Sex: %s, Age: %s\n", rowKey, name,sex,age);
        }
        // 关闭连接
        scanner.close();
        table.close();
        conn.close();
        str =true;
    }catch (Exception e){
            throw new RuntimeException(e);
        }return str;
    }
}
